package com.duoec.dw.lineage.service.analyser.statement;

import com.duoec.dw.lineage.dto.SqoopCommendArgument;
import com.duoec.dw.lineage.dto.statement.ImportSqoopStatement;
import org.springframework.stereotype.Service;

import java.util.List;

/**
 * @author xuwenzhen
 */
@Service
public class ImportSqoopStatementAnalyser extends BaseSqoopStatementAnalyser<ImportSqoopStatement> {
    private static final String PREFIX = "import";

    /**
     * 检查当前行是否需要当前分析器处理
     *
     * @param line 当前行
     * @return true，则会使用analyse方法解析
     */
    @Override
    public boolean check(String line) {
        String trimLine = line.trim();
        if (trimLine.startsWith(STR_SQOOP)) {
            line = trimLine.substring(STR_SQOOP.length()).trim();
        }
        return line.startsWith(PREFIX);
    }

    /**
     * 解析脚本，并返回当前的语句
     *
     * @param lines 所有行
     * @param index 开始行，0开始
     * @return 当前语句
     */
    @Override
    public ImportSqoopStatement analyse(List<String> lines, int index) {
        ImportSqoopStatement statement = new ImportSqoopStatement();
        //import --append --connect jdbc:mysql://10.50.10.44:3306/duoec_xfbp_broc?zeroDateTimeBehavior=round&tinyInt1isBit=false --driver com.mysql.jdbc.Driver --username esf_bigdata --password-file /user/oozie/share/lib/lib_20151229093730/password_mysql_pro.file  --table broc_estate_agent_dynamic --columns "dynamic_id,new_dynamic_id,estate_id,agent_id,staff_id,latest_action_type,latest_action_time,is_followed,is_visited,browse_num,call_num,share_num,news_interact_num,draw_coupon_num,referral_num,guide_num,purchase_num,startup_project_num,nugget_walk_num,nugget_train_num,watch_live_num,browse_num_in_30days,share_num_in_30days,draw_coupon_num_in_30days,referral_num_in_30days,guide_num_in_30days,purchase_num_in_30days,sort_score,is_deleted,create_time,update_time" --where "DATE_FORMAT(create_time,'%Y%m%d')=${dt}	or	DATE_FORMAT(update_time,'%Y%m%d')=${dt}" --target-dir /user/hive/warehouse/duoec_xfbp_broc_edw_source.db/s_du_broc_estate_agent_dynamic_delta/dt=${dt} --hive-drop-import-delims --input-fields-terminated-by \001 --fields-terminated-by \001 --skip-dist-cache
        String line = lines.get(index);
        SqoopCommendArgument args = getArgument(line);
        statement.setCommend(line);
        statement.setArgs(args);
        statement.setStart(index);
        statement.setEnd(index);
        return statement;
    }
}
